package com.enlorenz.core.channel.socket.common.server;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import com.enlorenz.common.util.SocketUtils;
import com.enlorenz.core.base.IReqMessageHandle;
import com.enlorenz.core.base.IReqResMessageHandle;
import com.enlorenz.core.channel.common.dstobj.IDstObjectManager;
import com.enlorenz.core.channel.common.dstobj.IDstObjectManagerGetter;
import com.enlorenz.core.channel.common.dstobj.IDstRoute;
import com.enlorenz.core.channel.common.message.IChannelSysMessage;
import com.enlorenz.core.channel.common.message.ILinkMessage;
import com.enlorenz.core.channel.socket.common.dstroute.IMessageRouteSender;
import com.enlorenz.core.channel.socket.common.dstroute.NioObjectManager;
import com.enlorenz.core.channel.socket.common.dstroute.category.NioDstRoute;
import com.enlorenz.core.channel.socket.common.reader.INioReader;
import com.enlorenz.core.channel.socket.common.translate.DataMessageTransformer;
import com.enlorenz.core.channel.socket.common.translate.IMessageBytesTransformer;
import com.enlorenz.core.channel.socket.common.util.RecvMessageOperator;
import com.enlorenz.core.channel.socket.common.writer.INioWriter;
import com.enlorenz.core.engine.EngineUtils;
import com.enlorenz.core.engine.exception.EngineControlException;
import com.enlorenz.core.engine.imp.thread.quick.AbstractQuickControlThread;
import com.enlorenz.core.queue.proc.AbstractObjectCycleThread;
import com.enlorenz.core.queue.transfer.imp.NoWaitMessageQueue;
import com.enlorenz.core.queue.transfer.imp.WaitMessageQueue;

/**
 * 长连接服务通道
 * 作为服务端，被动的接收其他的链接
 * 需要设置nioReader以及nioWriter以确定以何种方式读取或者写出
 * recvMsgHandle 是一项非常重要的设置参数，用于处理收到的数据信息
 * handRecvThreadNum 非常重要的处理线程数量参数，此参数直接影响到该服务端的处理性能
 * sysMsgHandle 用于接收系统消息处理器，比如写出成功等等
 * linkMsgHandle 通讯消息处理器，比如链接建立，链接断开等
 * @author lcb
 */
public abstract class AbstractServerChannel extends AbstractQuickControlThread implements 
	IMessageRouteSender,IDstObjectManagerGetter{
	private static final long serialVersionUID = 1L;
	//监听端口
	protected Integer listenPort;
	//接收外部传入的信息
	private WaitMessageQueue receiveQueue=new WaitMessageQueue();
	//通过连接发送消息
	protected NoWaitMessageQueue sendQueue=new NoWaitMessageQueue();
	//Nio读取者
	protected INioReader nioReader;
	//Nio写出者
	protected INioWriter nioWriter;
	//Socket地址
	private InetSocketAddress inetSocketAddress;
	//事件选择器
	private Selector selector;
	//服务端
	private ServerSocketChannel server;
	//公用缓存
	protected ByteBuffer commBuffer;
	//公用缓存大小
	private int commBufferSize=4096;
	//输入连接指示（对连接的请求）的最大队列长度被设置为 backlog 参数。如果队列满时收到连接指示，则拒绝该连接。
	private int backlog = 100;
	//消息转换器,如果没有传入值，则使用默认的转换器
	private IMessageBytesTransformer messageBytesTransformer
	=new DataMessageTransformer();
	//接收消息操作者
	private RecvMessageOperator recvMessageOperator
		=new RecvMessageOperator();
	//接收消息处理器
	private IReqResMessageHandle recvMsgHandle;
	//系统消息处理器
	private IReqMessageHandle sysMsgHandle;
	//系统通讯消息处理器
	private IReqResMessageHandle linkMsgHandle;
	//处理接收消息的线程列表
	private List<HandleThread> handleRecvThreadList=new ArrayList<HandleThread>();
	//处理接收线程的数量
	private int handRecvThreadNum=5;
	//是否初始化过
	private boolean isInit=false;
	//目标对象管理器
	private IDstObjectManager dstObjectManager=new NioObjectManager();
	//接收的客户限制数量
	private int acceptClientLimited=999999999;
	
	/**
	 * 发送消息
	 * @param message 消息
	 * @param dstObject 目标通道
	 */
	public void send(Object message,IDstRoute dstRoute){
		ServerInnerMessage sim=new ServerInnerMessage();
    	sim.setConnectedObject(dstRoute);
    	sim.setMessage(message);
    	sendQueue.addMsg(sim);
	}
	
	/**
	 * 初始化接收线程队列
	 */
	private void initRecvThreadList(){
		if(!isInit){
			for(int i=0;i<handRecvThreadNum;i++){
				HandleThread handleThread=new HandleThread();
				handleRecvThreadList.add(handleThread);
			}
			isInit=true;
		}
	}
	/**
	 * 校验输入参数
	 */
	private void validateInputParam(){
		if(null == listenPort){
			throw new RuntimeException("输入参数[listenPort{监听端口号}]不能为空");
		}
		
		if(null == recvMsgHandle){
			throw new RuntimeException("输入参数[recvMsgHandle{接收消息处理器}]不能为空");
		}
		
		if(null == nioReader){
			throw new RuntimeException("输入参数[nioReader{Nio读取者}]不能为空");
		}
		
		if(null == nioWriter){
			throw new RuntimeException("输入参数[nioWriter{Nio写出者}]不能为空");
		}
		
		logger.warn("请注意,当前的[{}:{}]服务端处理线程为[{}]个",
				new Object[]{this.getName(),this.listenPort,handRecvThreadNum});
	}
	
	/**
	 * 初始化Socket通讯地址
	 */
	protected void initEngine() throws EngineControlException {
		//校验输入参数
		validateInputParam();
		
		//初始化通讯地址
		inetSocketAddress = new InetSocketAddress(listenPort);
		//初始化公用缓存
		commBuffer = ByteBuffer.allocateDirect(commBufferSize);
		recvMessageOperator.setRecvQueue(receiveQueue);
		logger.info("[{}]初始化处理线程数量[{}]",this.getName(),handRecvThreadNum);
		
		//初始化接收线程队列
		initRecvThreadList();
		
		for(HandleThread handleRecvThread:handleRecvThreadList){
			//初始化接收线程
			handleRecvThread.init();
			handleRecvThread.setWaitMessageQueue(receiveQueue);
		}
		
		super.initEngine();
	}
	/**
	 * 启动Socket服务
	 */
	protected void startEngine() throws EngineControlException {
		try {
			//启动服务
			socketServerStart();
			//启动接收线程列表
			EngineUtils.startEngineList(handleRecvThreadList);
		} catch (IOException e) {
			throw new EngineControlException(this.getName(),e);
		}
		super.startEngine();
	}
	
	/**
	 * socket服务启动
	 * @throws IOException
	 */
	private void socketServerStart() throws IOException {
		try {
			//打开选择器
			selector = Selector.open();
			//打开通道   
			server = ServerSocketChannel.open(); 
			//非阻塞   
			server.configureBlocking(false); 
			//绑定地址
			server.socket().bind(inetSocketAddress, backlog);
			//向通道注册选择器和对应事件标识
			server.register(selector, SelectionKey.OP_ACCEPT); 
		} catch (IOException e) {
			close();
			throw e;
		}
	}
	
	/**
	 * 关闭打开的链接
	 */
	private void close() {
		//关闭选择器
		SocketUtils.closeSelector(selector);
		selector = null;
		//关闭Socket服务
		SocketUtils.closeServerSocketChannel(server);
		server = null;
		//关闭接收线程列表
		EngineUtils.stopEngineList(handleRecvThreadList);
	}
	
	/**
	 * 关闭Socket服务
	 */
	@Override
	protected void shutdownEngine() throws EngineControlException {
		//关闭所有接收的链接
		closeAllChannels();
		//关闭选择器和服务
		close();
		//调用父类的关闭
		super.shutdownEngine();
	}
	/**
	 * Socket是否还存活
	 */
	@Override
	public boolean isAlive() {
		//如果没有打开则没有链接上
		if(!SocketUtils.isSelectorOpen(selector)){
			return false;
		}
		//如果服务端没有打开
		if(!SocketUtils.isServerSocketChannelOpen(server)){
			return false;
		}
		
		return super.isAlive();
	}
	
	
	/**
	 * 加入需要输出的Key
	 */
	private void addWriteKey(){
		ServerInnerMessage message = (ServerInnerMessage) sendQueue.getMsg();
		if(null == message){
			return;
		}
		NioDstRoute nioObject=(NioDstRoute)message.getConnectedObject();
		SocketChannel channel=(SocketChannel)nioObject.getConnSource();
		Object srcMsg=(Object)message.getMessage();
		//如果链接则生成写事件
		if(SocketUtils.isChannelConnected(channel)){
			try{
				channel.register(selector,
						SelectionKey.OP_WRITE, srcMsg);
			}catch (ClosedChannelException e) {
				closeSocketChannel(channel,e);
			}
		//否则关闭链接	
		}else{
			closeSocketChannel(channel,new Exception("channel is not connected"));
		}
	}
	/**
	 * 关闭Socket通道
	 * @param channel
	 */
	protected void closeSocketChannel(SocketChannel channel,Exception ex){
		//关闭channel
		SocketUtils.closeChannel(channel);
		
		//移除内存对象
		NioDstRoute connObject=new NioDstRoute(channel);
		dstObjectManager.removeDstObject(connObject);
		try {
			sendMsg(channel,recvMessageOperator.createLinkDisconnMessage(connObject,ex));
		}catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	/**
	 * 关闭所有的链接
	 */
	private void closeAllChannels(){
		List<IDstRoute> scArr=dstObjectManager.removeAllDstObject();
		for(IDstRoute object:scArr){
			closeSocketChannel((SocketChannel)object.getConnSource(),new Exception("closeAllChannels"));
		}
	}
	
	/**
	 * 循环获取需要执行的Key,并执行相应的任务
	 * 
	 * @throws IOException
	 */
	@Override
	protected void handle() {
		// 加入写事件
		addWriteKey();
		// 参数10 ，防止多线程
		int nKeys = 0;
		try {
			if(SocketUtils.isSelectorOpen(selector)){
				nKeys = selector.select(10);
			}
		} catch (Exception e) {
			/*if (null!=selector&&
					selector.isOpen()) {
				// 发生异常，如果仍然开启，那么则返回继续
			} else {
				// 否则立即重启？
			}*/
			throw new RuntimeException(e);
		}
		
		// 如果没有Key,则返回
		if (0 == nKeys) {
			return;
		}
		
		Iterator<SelectionKey> it = selector.selectedKeys().iterator();
		while (it.hasNext()) {
			SelectionKey key = it.next();
			it.remove();
			// 判断Key是否合法
			if (!key.isValid()) {
				continue;
			}

			try {
				if (key.isAcceptable()) {
					handleAccept(key);
				} else if (key.isReadable()) {
					handleRead(key);
				} else if (key.isWritable()) {
					handleWrite(key);
				}
			} catch (Exception e) {
				// 如果是已经关闭异常
				if (e instanceof ClosedChannelException
						|| e instanceof IOException) {
					closeSocketChannel((SocketChannel) key.channel(),e);
					key.cancel();
				}
				e.printStackTrace();
				throw new RuntimeException(e);
			}
		}

	}
    
    
    /**
     * 封装并发送信息
     * @param requestMessage
     * @param channel
     */
    protected void sendWrapMessage(SocketChannel channel,byte [] requstMsg){
    	//将字节流转为消息
    	Object message=messageBytesTransformer.bytes2Message(requstMsg);
    	sendMsg(channel,message);
    }
    
    /**
     * 发送消息
     * @param message
     */
    protected void sendMsg(SocketChannel channel,Object message){
    	ServerInnerMessage sim=new ServerInnerMessage();
    	sim.setConnectedObject(new NioDstRoute(channel));
    	sim.setMessage(message);
    	recvMessageOperator.sendRecvMessage(sim);
    }
    
    /**
     * 发送消息
     * @param message
     */
    protected void sendMsg(Object message,IDstRoute dstObject){
    	ServerInnerMessage sim=new ServerInnerMessage();
    	sim.setConnectedObject(dstObject);
    	sim.setMessage(message);
    	recvMessageOperator.sendRecvMessage(sim);
    }
    
    
    /**
    /**
	 * 处理链接建立
	 * @param key
	 * @throws Exception
	 */
    protected void handleAccept(SelectionKey key) throws Exception {  
        ServerSocketChannel server = (ServerSocketChannel)key.channel(); 
        //接收链接
        SocketChannel channel = server.accept();   
        //大于限制的链接数量,必须大于等于
        if(dstObjectManager.size()>=getAcceptClientLimited()){
        	SocketUtils.closeChannel(channel);
        	throw new RuntimeException("异常!客户端连接数量超过:"+getAcceptClientLimited());
        }
        
        //设置为非阻塞模式
        channel.configureBlocking(false);
        //注册读事件
        channel.register(key.selector(), SelectionKey.OP_READ);
        NioDstRoute nioConnObject=new NioDstRoute(channel);
        //加入集合
        dstObjectManager.addDstObject(nioConnObject);
        //发送链接接收消息
        sendMsg(channel,recvMessageOperator.createLinkConnMessage(nioConnObject));
    }
    
    /**
     * 处理读取
     * @param key
     * @throws Exception
     */
    protected void handleRead(SelectionKey key) throws Exception {  
    	SocketChannel channel = (SocketChannel)key.channel();
    	nioReader.setChannel(channel);
    	nioReader.setCommBuffer(commBuffer);

		
    	//读取请求信息
    	byte [] requestMsg=nioReader.read();
    	if(null!=requestMsg){
    		//封装并发送信息
    		sendWrapMessage(channel,requestMsg);
    	}
    	doAfterRead(key);
    }  
   /**
    * 在读取之后需要操作的事情
    * @param key
    */
    protected abstract void doAfterRead(SelectionKey key);
    
    /**
     * 处理写
     * @param key
     * @throws Exception
     */
    protected void handleWrite(SelectionKey key) throws Exception {  
    	Object responseMessage=(Object)key.attachment();
    	SocketChannel channel = (SocketChannel)key.channel(); 
    	try{
	    	nioWriter.setSocketChannel(channel);
	    	//将消息转化为字节流
	    	byte [] data=messageBytesTransformer.message2Bytes(responseMessage);
	    	nioWriter.write(data);
	    	doAfterWrite(key);
	    	//发送成功消息
	    	
	    	sendMsg(channel,recvMessageOperator.createWriteStatusMessage(true,responseMessage,null));
    	}catch (Exception e) {
    		//发送异常消息
    		sendMsg(channel,recvMessageOperator.createWriteStatusMessage(false,responseMessage,e));
    		throw e;
		}
    }
    
    /**
     * 处理接收消息
     * @author lcb
     */
    private class HandleThread extends AbstractObjectCycleThread{
		@Override
		protected void process(Object result) {
			
			ServerInnerMessage servMsg=(ServerInnerMessage)result;
			try {
				Object srcMsg=servMsg.getMessage();
				//对消息进行处理
				Object resultMsg=null;
				//如果判断出是系统消息，系统消息不需要返回
				if(srcMsg instanceof IChannelSysMessage){
					//不一定需要设置处理器
					if(null!=sysMsgHandle){
						sysMsgHandle.reqMsgHandle(srcMsg);
					}
				//如果判断是通讯消息
				}else if(srcMsg instanceof ILinkMessage){
					//不一定需要设置处理器
					if(null!=linkMsgHandle){
						resultMsg=linkMsgHandle.reqResMsgHandle(srcMsg);
					}
				//如果是其他的数据消息
				}else{
					//必须设置处理器，不然报错
					resultMsg=recvMsgHandle.reqResMsgHandle(srcMsg);
				}
				if(null!=resultMsg){
					servMsg.setMessage(resultMsg);
					sendQueue.addMsg(servMsg);
				}
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
    }
    
    /**
     * 在写之后需要操作的事情
     * @param key
     */
    protected abstract void doAfterWrite(SelectionKey key);
    
    /**
	 * 接收处理线程数量
	 * 此参数非常重要，直接引擎接收消息的处理效率
	 * @param handRecvThreadNum 接收处理线程数量
	 */
	public void setHandRecvThreadNum(int handRecvThreadNum) {
		this.handRecvThreadNum = handRecvThreadNum;
	}
	public IDstObjectManager getDstObjectManager() {
		return dstObjectManager;
	}

	public void setSysMsgHandle(IReqMessageHandle sysMsgHandle) {
		this.sysMsgHandle = sysMsgHandle;
	}

	public void setHandleRecvThreadList(List<HandleThread> handleRecvThreadList) {
		this.handleRecvThreadList = handleRecvThreadList;
	}
	
	public void setLinkMsgHandle(IReqResMessageHandle linkMsgHandle) {
		this.linkMsgHandle = linkMsgHandle;
	}
	public void setDstObjectManager(IDstObjectManager dstObjectManager) {
		this.dstObjectManager = dstObjectManager;
	}
	public void setRecvMsgHandle(IReqResMessageHandle recvMsgHandle) {
		this.recvMsgHandle = recvMsgHandle;
	}
	public void setMessageBytesTransformer(
			IMessageBytesTransformer messageBytesTransformer) {
		this.messageBytesTransformer = messageBytesTransformer;
	}
	public void setListenPort(Integer listenPort) {
		this.listenPort = listenPort;
	}
	public void setCommBufferSize(int commBufferSize) {
		this.commBufferSize = commBufferSize;
	}
	public void setBacklog(int backlog) {
		this.backlog = backlog;
	}
	public void setNioWriter(INioWriter nioWriter) {
		this.nioWriter = nioWriter;
	}
	
	public void setNioReader(INioReader nioReader) {
		this.nioReader = nioReader;
	}

	public int getAcceptClientLimited() {
		return acceptClientLimited;
	}

	public void setAcceptClientLimited(int acceptClientLimited) {
		this.acceptClientLimited = acceptClientLimited;
	}
	
}
